from pyflink.common import SimpleStringSchema, WatermarkStrategy
from pyflink.datastream import StreamExecutionEnvironment, KeyedProcessFunction
from pyflink.datastream.connectors.kafka import KafkaSource, KafkaOffsetsInitializer

env = StreamExecutionEnvironment.get_execution_environment()

env.set_parallelism(1)

source = KafkaSource.builder() \
    .set_bootstrap_servers("master:9092") \
    .set_topics("fraud") \
    .set_group_id("my-group") \
    .set_starting_offsets(KafkaOffsetsInitializer.latest()) \
    .set_value_only_deserializer(SimpleStringSchema()) \
    .build()

lines_ds = env.from_source(source, WatermarkStrategy.no_watermarks(), "Kafka Source")

faruds_ds = lines_ds.map(lambda line: (line.split(",")[0], float(line.split(",")[1])))

key_by_ds = faruds_ds.key_by(lambda x: x[0])

class FraudKeyedProcessFunction(KeyedProcessFunction):
    def __init__(self):
        self.flag = dict()
        self.min_price = dict()

    def process_element(self, value, ctx: 'KeyedProcessFunction.Context'):

         user_id = value[0]
         price = value[1]


         if self.flag.get(user_id):
             if price > 500:
                 print('发现异常交易')
                 min_price = self.min_price[user_id]
                 yield user_id, price, min_price
             self.flag[user_id] = False
             del self.min_price[user_id]

         if price < 1:
             print('出现小于1的交易')
             self.flag[user_id] = True
             self.min_price[user_id] = price


alter_ds = key_by_ds.process(FraudKeyedProcessFunction())

alter_ds.print()

env.execute()


'''
1,0.1
1,20
1,0.6
1,560
2,58
2,0.8
2,880

'''